//! 配置广播功能集成测试
//! 
//! 该文件包含配置广播功能的完整单元测试和Mock测试

#[cfg(feature = "broadcast")]
use rustcloud_config::broadcast::*;
#[cfg(feature = "broadcast")]
use rustcloud_config::bus_broadcast::*;
use rustcloud_config::error::{ConfigResult, ConfigError};

use std::collections::HashMap;
use std::sync::Arc;
use std::time::SystemTime;
use serde_json::{json, Value};
use tokio;

// ============================================================================
// ConfigBroadcastEventType 集成测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod config_broadcast_event_type_integration_tests {
    use super::*;

    #[test]
    fn test_config_broadcast_event_type_as_str() {
        // 测试所有事件类型的字符串表示
        assert_eq!(ConfigBroadcastEventType::ConfigUpdate.as_str(), "config.update");
        assert_eq!(ConfigBroadcastEventType::ConfigDelete.as_str(), "config.delete");
        assert_eq!(ConfigBroadcastEventType::ConfigRefresh.as_str(), "config.refresh");
        assert_eq!(ConfigBroadcastEventType::ServiceRestart.as_str(), "service.restart");
    }

    #[test]
    fn test_config_broadcast_event_type_equality() {
        // 测试事件类型相等性
        assert_eq!(ConfigBroadcastEventType::ConfigUpdate, ConfigBroadcastEventType::ConfigUpdate);
        assert_ne!(ConfigBroadcastEventType::ConfigUpdate, ConfigBroadcastEventType::ConfigDelete);
        assert_ne!(ConfigBroadcastEventType::ConfigRefresh, ConfigBroadcastEventType::ServiceRestart);
    }

    #[test]
    fn test_config_broadcast_event_type_clone() {
        // 测试克隆功能
        let event_type = ConfigBroadcastEventType::ConfigUpdate;
        let cloned = event_type.clone();
        assert_eq!(event_type, cloned);
        
        let event_type2 = ConfigBroadcastEventType::ServiceRestart;
        let cloned2 = event_type2.clone();
        assert_eq!(event_type2, cloned2);
    }

    #[test]
    fn test_config_broadcast_event_type_debug() {
        // 测试Debug输出
        let event_type = ConfigBroadcastEventType::ConfigUpdate;
        let debug_str = format!("{:?}", event_type);
        assert!(debug_str.contains("ConfigUpdate"));
        
        let event_type2 = ConfigBroadcastEventType::ConfigDelete;
        let debug_str2 = format!("{:?}", event_type2);
        assert!(debug_str2.contains("ConfigDelete"));
    }

    #[test]
    fn test_config_broadcast_event_type_serialization() {
        // 测试序列化
        let event_type = ConfigBroadcastEventType::ConfigUpdate;
        let serialized = serde_json::to_string(&event_type).unwrap();
        assert!(serialized.contains("ConfigUpdate"));
        
        // 测试反序列化
        let deserialized: ConfigBroadcastEventType = serde_json::from_str(&serialized).unwrap();
        assert_eq!(event_type, deserialized);
        
        // 测试所有类型
        let all_types = vec![
            ConfigBroadcastEventType::ConfigUpdate,
            ConfigBroadcastEventType::ConfigDelete,
            ConfigBroadcastEventType::ConfigRefresh,
            ConfigBroadcastEventType::ServiceRestart,
        ];
        
        for event_type in all_types {
            let serialized = serde_json::to_string(&event_type).unwrap();
            let deserialized: ConfigBroadcastEventType = serde_json::from_str(&serialized).unwrap();
            assert_eq!(event_type, deserialized);
        }
    }
}

// ============================================================================
// ConfigBroadcastEvent 集成测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod config_broadcast_event_integration_tests {
    use super::*;

    #[test]
    fn test_config_broadcast_event_config_update_creation() {
        // 测试配置更新事件创建 - 完整参数
        let event = ConfigBroadcastEvent::config_update(
            "user-service",
            "production",
            Some("v1.0"),
            "database.url",
            Some(json!("old-url")),
            json!("new-url"),
            "config-server",
            Some("admin"),
        );

        assert_eq!(event.application, "user-service");
        assert_eq!(event.profile, "production");
        assert_eq!(event.label, Some("v1.0".to_string()));
        assert_eq!(event.config_path, Some("database.url".to_string()));
        assert_eq!(event.old_value, Some(json!("old-url")));
        assert_eq!(event.new_value, Some(json!("new-url")));
        assert_eq!(event.source_service, "config-server");
        assert_eq!(event.user, Some("admin".to_string()));
        assert_eq!(event.event_type, ConfigBroadcastEventType::ConfigUpdate);
        assert!(event.id.starts_with("config_update_"));
        assert!(event.target_services.is_none());
        
        // 检查时间戳是最近的
        let now = SystemTime::now();
        let duration = now.duration_since(event.timestamp).unwrap_or_default();
        assert!(duration.as_secs() < 5); // 应该在5秒内创建
    }

    #[test]
    fn test_config_broadcast_event_service_restart_creation() {
        // 测试服务重启事件创建 - 有目标服务
        let target_services = vec!["service1".to_string(), "service2".to_string()];
        let event = ConfigBroadcastEvent::service_restart(
            "app",
            "source-service",
            Some(target_services.clone()),
            Some("operator"),
        );

        assert_eq!(event.application, "app");
        assert_eq!(event.profile, "");
        assert_eq!(event.source_service, "source-service");
        assert_eq!(event.target_services, Some(target_services));
        assert_eq!(event.user, Some("operator".to_string()));
        assert_eq!(event.event_type, ConfigBroadcastEventType::ServiceRestart);
        assert!(event.id.starts_with("service_restart_"));
        assert_eq!(event.config_path, None);
        assert_eq!(event.old_value, None);
        assert_eq!(event.new_value, None);
    }

    #[test]
    fn test_config_broadcast_event_config_refresh_creation() {
        // 测试配置刷新事件创建
        let targets = vec!["target1".to_string()];
        let event = ConfigBroadcastEvent::config_refresh(
            "application",
            "staging",
            "config-server",
            Some(targets.clone()),
        );

        assert_eq!(event.application, "application");
        assert_eq!(event.profile, "staging");
        assert_eq!(event.source_service, "config-server");
        assert_eq!(event.target_services, Some(targets));
        assert_eq!(event.event_type, ConfigBroadcastEventType::ConfigRefresh);
        assert!(event.id.starts_with("config_refresh_"));
        assert_eq!(event.user, None);
        assert_eq!(event.config_path, None);
        assert_eq!(event.old_value, None);
        assert_eq!(event.new_value, None);
    }

    #[test]
    fn test_config_broadcast_event_unicode_strings() {
        // 测试Unicode字符串
        let event = ConfigBroadcastEvent::config_update(
            "用户服务",
            "生产环境",
            Some("版本1.0"),
            "数据库.地址",
            Some(json!("旧地址")),
            json!("新地址"),
            "配置服务器",
            Some("管理员"),
        );

        assert_eq!(event.application, "用户服务");
        assert_eq!(event.profile, "生产环境");
        assert_eq!(event.label, Some("版本1.0".to_string()));
        assert_eq!(event.config_path, Some("数据库.地址".to_string()));
        assert_eq!(event.source_service, "配置服务器");
        assert_eq!(event.user, Some("管理员".to_string()));
    }

    #[test]
    fn test_config_broadcast_event_serialization() {
        // 测试事件序列化和反序列化
        let event = ConfigBroadcastEvent::config_update(
            "test-service",
            "test",
            None,
            "test.key",
            None,
            json!({"nested": {"key": "value"}}),
            "source",
            None,
        );

        let serialized = serde_json::to_string(&event).unwrap();
        let deserialized: ConfigBroadcastEvent = serde_json::from_str(&serialized).unwrap();

        assert_eq!(event.application, deserialized.application);
        assert_eq!(event.profile, deserialized.profile);
        assert_eq!(event.event_type, deserialized.event_type);
        assert_eq!(event.source_service, deserialized.source_service);
        assert_eq!(event.new_value, deserialized.new_value);
    }
}

// ============================================================================
// ConfigMessage 集成测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod config_message_integration_tests {
    use super::*;

    #[test]
    fn test_config_message_creation() {
        let payload = json!({"test": "data", "number": 42});
        let message = ConfigMessage::new(
            "test.topic".to_string(),
            payload.clone(),
        );

        assert_eq!(message.topic, "test.topic");
        assert_eq!(message.payload, payload);
        assert!(message.id.starts_with("msg_"));
        assert!(message.headers.is_empty());
        assert!(message.correlation_id.is_none());
        
        // 检查时间戳是最近的
        let now = SystemTime::now();
        let duration = now.duration_since(message.timestamp).unwrap_or_default();
        assert!(duration.as_secs() < 1);
    }

    #[test]
    fn test_config_message_with_header() {
        let message = ConfigMessage::new(
            "topic".to_string(),
            json!({}),
        ).with_header("key1".to_string(), "value1".to_string())
         .with_header("key2".to_string(), "value2".to_string())
         .with_header("Content-Type".to_string(), "application/json".to_string());

        assert_eq!(message.headers.len(), 3);
        assert_eq!(message.headers.get("key1"), Some(&"value1".to_string()));
        assert_eq!(message.headers.get("key2"), Some(&"value2".to_string()));
        assert_eq!(message.headers.get("Content-Type"), Some(&"application/json".to_string()));
    }

    #[test]
    fn test_config_message_chaining() {
        // 测试方法链式调用
        let message = ConfigMessage::new(
            "test.topic".to_string(),
            json!({"data": "test"}),
        )
        .with_header("type".to_string(), "config".to_string())
        .with_header("source".to_string(), "server".to_string())
        .with_header("priority".to_string(), "high".to_string())
        .with_correlation_id("corr-123".to_string());

        assert_eq!(message.topic, "test.topic");
        assert_eq!(message.headers.len(), 3);
        assert_eq!(message.correlation_id, Some("corr-123".to_string()));
        assert_eq!(message.headers.get("type"), Some(&"config".to_string()));
        assert_eq!(message.headers.get("source"), Some(&"server".to_string()));
        assert_eq!(message.headers.get("priority"), Some(&"high".to_string()));
    }

    #[test]
    fn test_config_message_complex_payload() {
        let complex_payload = json!({
            "config": {
                "database": {
                    "host": "localhost",
                    "port": 5432,
                    "credentials": {
                        "username": "admin",
                        "password": "secret"
                    }
                },
                "features": ["auth", "cache", "monitoring"],
                "settings": {
                    "timeout": 30,
                    "retries": 3
                }
            },
            "metadata": {
                "version": "1.0.0",
                "environment": "production",
                "timestamp": "2024-01-01T00:00:00Z"
            }
        });

        let message = ConfigMessage::new(
            "config.complex".to_string(),
            complex_payload.clone(),
        );

        assert_eq!(message.payload, complex_payload);
        assert_eq!(message.topic, "config.complex");
    }
}

// ============================================================================
// InMemoryConfigMessageBus 集成测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod in_memory_message_bus_integration_tests {
    use super::*;

    #[tokio::test]
    async fn test_in_memory_message_bus_creation() {
        let bus = InMemoryConfigMessageBus::new();
        
        // 测试创建发布者
        let publisher_result = bus.create_publisher().await;
        assert!(publisher_result.is_ok());
        
        // 测试创建订阅者
        let subscriber_result = bus.create_subscriber().await;
        assert!(subscriber_result.is_ok());
    }

    #[tokio::test]
    async fn test_in_memory_message_bus_publish_subscribe() {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        
        // 创建发布者和订阅者
        let publisher = bus.create_publisher().await.unwrap();
        let subscriber = bus.create_subscriber().await.unwrap();
        
        // 订阅主题
        subscriber.subscribe("test.topic").await.unwrap();
        
        // 发布消息
        let message = ConfigMessage::new(
            "test.topic.specific".to_string(),
            json!({"data": "test", "id": 123}),
        );
        publisher.publish(message.clone()).await.unwrap();
        
        // 接收消息
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_some());
        
        let received_msg = received.unwrap();
        assert_eq!(received_msg.topic, message.topic);
        assert_eq!(received_msg.payload, message.payload);
    }

    #[tokio::test]
    async fn test_in_memory_message_bus_multiple_messages() {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        let publisher = bus.create_publisher().await.unwrap();
        let subscriber = bus.create_subscriber().await.unwrap();
        
        subscriber.subscribe("test").await.unwrap();
        
        // 发布多条消息
        for i in 0..5 {
            let message = ConfigMessage::new(
                format!("test.message.{}", i),
                json!({"index": i, "data": format!("message_{}", i)}),
            );
            publisher.publish(message).await.unwrap();
        }
        
        // 接收所有消息
        let mut received_count = 0;
        while let Ok(Some(_)) = subscriber.receive().await {
            received_count += 1;
        }
        
        assert_eq!(received_count, 5);
    }

    #[tokio::test]
    async fn test_in_memory_message_bus_topic_filtering() {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        let publisher = bus.create_publisher().await.unwrap();
        let subscriber = bus.create_subscriber().await.unwrap();
        
        // 只订阅特定主题
        subscriber.subscribe("config.update").await.unwrap();
        
        // 发布不同主题的消息
        let config_message = ConfigMessage::new(
            "config.update.database".to_string(),
            json!({"type": "config", "key": "database.url"}),
        );
        let other_message = ConfigMessage::new(
            "system.notification".to_string(),
            json!({"type": "notification", "message": "system update"}),
        );
        
        publisher.publish(config_message).await.unwrap();
        publisher.publish(other_message).await.unwrap();
        
        // 应该只收到匹配的消息
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_some());
        assert!(received.unwrap().topic.starts_with("config.update"));
        
        // 第二次调用应该没有更多消息（other_message被过滤掉了）
        let no_more = subscriber.receive().await.unwrap();
        assert!(no_more.is_none());
    }

    #[tokio::test]
    async fn test_in_memory_message_bus_unsubscribe() {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        let publisher = bus.create_publisher().await.unwrap();
        let subscriber = bus.create_subscriber().await.unwrap();
        
        subscriber.subscribe("test.topic").await.unwrap();
        
        // 发布消息前取消订阅
        subscriber.unsubscribe("test.topic").await.unwrap();
        
        let message = ConfigMessage::new(
            "test.topic.message".to_string(),
            json!({"data": "test"}),
        );
        publisher.publish(message).await.unwrap();
        
        // 应该收不到消息
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_none());
    }
}

// ============================================================================
// AbstractConfigBroadcastManager 集成测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod abstract_config_broadcast_manager_integration_tests {
    use super::*;

    fn create_test_manager() -> AbstractConfigBroadcastManager {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        AbstractConfigBroadcastManager::new(
            bus,
            "test-service".to_string(),
            Some("test.broadcast".to_string()),
        )
    }

    #[tokio::test]
    async fn test_abstract_config_broadcast_manager_creation() {
        let manager = create_test_manager();
        
        // 测试初始状态
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 0);
        assert_eq!(stats.events_received, 0);
        assert_eq!(stats.active_listeners, 0);
        
        let cluster_state = manager.get_cluster_state().await;
        assert!(cluster_state.is_empty());
    }

    #[tokio::test]
    async fn test_broadcast_config_update() {
        let manager = create_test_manager();
        
        // 测试配置更新广播 - 完整参数
        let result = manager.broadcast_config_update(
            "user-service",
            "production",
            Some("v1.0"),
            "database.url",
            Some(json!("old-url")),
            json!("new-url"),
            Some("admin"),
        ).await;
        
        assert!(result.is_ok());
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 1);
    }

    #[tokio::test]
    async fn test_broadcast_config_delete() {
        let manager = create_test_manager();
        
        // 测试配置删除广播 - 完整参数
        let result = manager.broadcast_config_delete(
            "user-service",
            "production",
            Some("v1.0"),
            "database.password",
            Some("admin"),
        ).await;
        
        assert!(result.is_ok());
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 1);
    }

    #[tokio::test]
    async fn test_broadcast_config_refresh() {
        let manager = create_test_manager();
        
        // 测试配置刷新广播 - 有目标服务
        let target_services = vec!["service1".to_string(), "service2".to_string()];
        let result = manager.broadcast_config_refresh(
            "application",
            "production",
            Some(target_services),
        ).await;
        
        assert!(result.is_ok());
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 1);
    }

    #[tokio::test]
    async fn test_broadcast_service_restart() {
        let manager = create_test_manager();
        
        // 测试服务重启广播 - 有目标服务和用户
        let target_services = vec!["target-service".to_string()];
        let result = manager.broadcast_service_restart(
            "application",
            Some(target_services),
            Some("operator"),
        ).await;
        
        assert!(result.is_ok());
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 1);
    }

    #[tokio::test]
    async fn test_multiple_broadcasts() {
        let manager = create_test_manager();
        
        // 发送多个不同类型的广播
        manager.broadcast_config_update(
            "service1", "prod", None, "key1", None, json!("value1"), None
        ).await.unwrap();
        
        manager.broadcast_config_delete(
            "service2", "dev", None, "key2", None
        ).await.unwrap();
        
        manager.broadcast_config_refresh(
            "service3", "test", None
        ).await.unwrap();
        
        manager.broadcast_service_restart(
            "service4", None, None
        ).await.unwrap();
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 4);
    }

    #[tokio::test]
    async fn test_unicode_parameters() {
        let manager = create_test_manager();
        
        // 测试Unicode参数
        let result = manager.broadcast_config_update(
            "用户服务",
            "生产环境",
            Some("版本1.0"),
            "数据库.配置.地址",
            Some(json!("旧地址")),
            json!("新地址"),
            Some("管理员"),
        ).await;
        
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_large_payload() {
        let manager = create_test_manager();
        
        // 创建大型payload
        let mut large_config = serde_json::Map::new();
        for i in 0..100 {
            large_config.insert(
                format!("config_key_{}", i),
                json!(format!("config_value_{}_with_content", i))
            );
        }
        
        let result = manager.broadcast_config_update(
            "service",
            "prod",
            None,
            "large.config",
            None,
            json!(large_config),
            None,
        ).await;
        
        assert!(result.is_ok());
    }
}

// ============================================================================
// 边界条件和错误处理测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod boundary_and_error_integration_tests {
    use super::*;

    fn create_test_manager() -> AbstractConfigBroadcastManager {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        AbstractConfigBroadcastManager::new(
            bus,
            "test-service".to_string(),
            None,
        )
    }

    #[tokio::test]
    async fn test_empty_strings() {
        let manager = create_test_manager();
        
        // 测试空字符串参数
        let result = manager.broadcast_config_update(
            "",
            "",
            Some(""),
            "",
            None,
            json!(""),
            Some(""),
        ).await;
        
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_special_characters() {
        let manager = create_test_manager();
        
        let special_chars = "!@#$%^&*()_+-=[]{}|;':,.<>?/~`";
        
        let result = manager.broadcast_config_update(
            special_chars,
            special_chars,
            Some(special_chars),
            special_chars,
            None,
            json!(special_chars),
            Some(special_chars),
        ).await;
        
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_null_and_empty_json_values() {
        let manager = create_test_manager();
        
        // 测试null值
        let result1 = manager.broadcast_config_update(
            "service", "prod", None, "key", None, json!(null), None
        ).await;
        assert!(result1.is_ok());
        
        // 测试空对象
        let result2 = manager.broadcast_config_update(
            "service", "prod", None, "key", None, json!({}), None
        ).await;
        assert!(result2.is_ok());
        
        // 测试空数组
        let result3 = manager.broadcast_config_update(
            "service", "prod", None, "key", None, json!([]), None
        ).await;
        assert!(result3.is_ok());
    }

    #[tokio::test]
    async fn test_concurrent_operations() {
        let manager = Arc::new(create_test_manager());
        
        // 并发执行多个操作
        let mut handles = vec![];
        
        for i in 0..10 {
            let manager_clone = manager.clone();
            let handle = tokio::spawn(async move {
                manager_clone.broadcast_config_update(
                    &format!("service-{}", i),
                    "prod",
                    None,
                    "key",
                    None,
                    json!(format!("value-{}", i)),
                    None,
                ).await
            });
            handles.push(handle);
        }
        
        // 等待所有操作完成
        for handle in handles {
            let result = handle.await.unwrap();
            assert!(result.is_ok());
        }
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 10);
    }
}